package spring.cloud.tasks.executor.listener.failover;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.executor.node.Node;
import spring.cloud.tasks.executor.utils.execute_callback.LeaderExecuteCallback;
import spring.cloud.tasks.executor.utils.log.LogUtils;

import java.util.List;

@Slf4j
@Setter
@Getter
class FailoverLeaderExecuteCallback implements LeaderExecuteCallback {

    private final FailoverService failoverService;

    public FailoverLeaderExecuteCallback(FailoverService failoverService) {
        this.failoverService = failoverService;
    }

    @Override
    public void execute() {
        if (!failoverService.needFailover()) {
            return;
        }
        if (failoverService.taskExecutor == null) {
            return;
        }
        if (failoverService.zookeeper.checkExists(Node.getExecutorLostTrafficNodePath(failoverService.executorId))) {
            return;
        }
        if (!failoverService.taskExecutor.getTaskConfigurationService().getPreferExecutorIdList().contains(failoverService.executorId) && !failoverService.taskExecutor.getTaskConfigurationService().isUseDispreferList()) {
            return;
        }
        List<String> items = failoverService.getTaskNodeStorage().getTaskNodeChildrenKeys(FailoverNode.ITEMS_ROOT);
        if (items != null && !items.isEmpty()) {
            int crashedItem = Integer.parseInt(failoverService.getTaskNodeStorage().getTaskNodeChildrenKeys(FailoverNode.ITEMS_ROOT).get(0));
            LogUtils.debug(log, failoverService.taskId, "Elastic job: failover job begin, crashed item:{}.", crashedItem);
            failoverService.getTaskNodeStorage().fillEphemeralTaskNode(FailoverNode.getExecutionFailoverNode(crashedItem), failoverService.executorId);
            failoverService.getTaskNodeStorage().removeTaskNodeIfExisted(FailoverNode.getItemsNode(crashedItem));
            failoverService.taskExecutor.triggerTask();
        }
    }
}
